package hadoop.mapReduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

/**
 * @describe:  自定义key 排序
 */
public class SlorBeanKey {

    //2020-04-14 16:34:03.800 [main] INFO
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException {
        System.setProperty("HADOOP_USER_NAME", "root");
        String HDFS_URI = "hdfs://hadoop1:9000/";
        String USER = "root";
        Configuration conf = new Configuration();
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        conf.set("dfs.client.use.datanode.hostname", "true");
        conf.set("dfs.replication", "1");
        FileSystem fs = FileSystem.get(new URI(HDFS_URI), conf, USER);
        Job job = Job.getInstance(conf);

        //job所在的jar包
        job.setJarByClass(SlorBeanKey.class);

        job.setMapperClass(SlorBeanKeyMapper.class);
        job.setReducerClass(SlorBeanKeyReducer.class);

        job.setMapOutputKeyClass(Bean.class);
        job.setMapOutputValueClass(Text.class);

        job.setOutputKeyClass(Bean.class);
        job.setOutputValueClass(Text.class);

        //分区数需要和任务数匹配
        job.setNumReduceTasks(1);

        String inputPath = "/user/root/input";
        String outputPath = "/user/root/output";

        Path path = new Path(outputPath);
        FileSystem fileSystem = path.getFileSystem(conf);
        if (fileSystem.exists(path)) {
            fileSystem.delete(path, true);
        }
        //设置本次job的输入 输出
        FileInputFormat.setInputPaths(job, new Path(inputPath));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        //提交job
        job.waitForCompletion(true);
    }

    static class SlorBeanKeyMapper extends Mapper<LongWritable, Text, Bean, Text> {
        Bean tx = new Bean();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] fields = line.split(" ");
            tx.set(fields[0], fields[1]);
            context.write(tx, value);
        }
    }

    static class SlorBeanKeyReducer extends Reducer<Bean, Text, Bean, Text> {
        @Override
        protected void reduce(Bean key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            for (Text bean : values) {
                context.write(key, bean);
            }
        }
    }


    static class Bean implements WritableComparable<Bean> {
        private String date;
        private String time;

        public Bean() {
        }

        public Bean(String date, String time) {
            set(date, time);
        }

        public void set(String date, String time) {
            this.date = date;
            this.time = time;
        }

        @Override
        public void write(DataOutput out) throws IOException {
            out.writeUTF(date);
            out.writeUTF(time);
        }

        @Override
        public void readFields(DataInput in) throws IOException {
            date = in.readUTF();
            time = in.readUTF();
        }

        @Override
        public String toString() {
            return date + "\t" + time;
        }

        @Override
        public int compareTo(Bean key) {
            String a = date + time;
            String b = key.date + key.time;
            return a.compareTo(b);
        }
    }

}
